package org.mq.mymq.common.remote;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.mq.mymq.common.cache.BrokerServerSyncFutureManager;
import org.mq.mymq.common.coder.TcpMsg;
import org.mq.mymq.common.dto.ConsumeMsgAckRespDTO;
import org.mq.mymq.common.dto.ConsumeMsgBaseRespDTO;
import org.mq.mymq.common.dto.SendMessageToBrokerResponseDTO;
import org.mq.mymq.common.enums.BrokerResponseCode;
import org.mq.mymq.common.event.EventBus;

/**
 * @ClassName BrokerRemoteRespHandler
 * @Description
 * @Author jiarong_ye
 * @Date 2025/6/27 14:17
 * @Version 1.0
 */
@ChannelHandler.Sharable
public class BrokerRemoteRespHandler extends SimpleChannelInboundHandler<TcpMsg> {
    private EventBus eventBus;

    public BrokerRemoteRespHandler(EventBus eventBus) {
        this.eventBus = eventBus;
        this.eventBus.init();
    }
    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        if (BrokerResponseCode.SEND_MSG_RESP.getCode() == code) {
            SendMessageToBrokerResponseDTO sendMessageToBrokerResponseDTO = JSON.parseObject(body, SendMessageToBrokerResponseDTO.class);
            SyncFuture syncFuture = BrokerServerSyncFutureManager.get(sendMessageToBrokerResponseDTO.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(tcpMsg);
            }
        } else if (BrokerResponseCode.CONSUME_MSG_RESP.getCode() == code) {
            ConsumeMsgBaseRespDTO consumeMsgBaseRespDTO = JSON.parseObject(body, ConsumeMsgBaseRespDTO.class);
            SyncFuture syncFuture = BrokerServerSyncFutureManager.get(consumeMsgBaseRespDTO.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(tcpMsg);
            }
        } else if (BrokerResponseCode.BROKER_UPDATE_OFFSET_RESP.getCode() == code) {
            ConsumeMsgAckRespDTO consumeMsgAckRespDTO = JSON.parseObject(body, ConsumeMsgAckRespDTO.class);
            SyncFuture syncFuture = BrokerServerSyncFutureManager.get(consumeMsgAckRespDTO.getMsgId());
            if (syncFuture != null) {
                syncFuture.setResponse(tcpMsg);
            }
        }
    }
}
